package org.jetlinks.pro.timeseries.micrometer;

import io.micrometer.core.instrument.Clock;
import io.micrometer.core.instrument.step.StepMeterRegistry;
import io.micrometer.core.instrument.util.NamedThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.jetlinks.core.metadata.DataType;
import org.jetlinks.core.metadata.types.StringType;
import org.jetlinks.pro.timeseries.TimeSeriesManager;
import org.jetlinks.pro.timeseries.TimeSeriesMetric;
import reactor.core.publisher.Flux;

import javax.annotation.Nonnull;
import java.util.*;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
public class TimeSeriesMeterRegistry extends StepMeterRegistry {

    TimeSeriesManager timeSeriesManager;

    TimeSeriesMetric metric;

    private static final ThreadFactory DEFAULT_THREAD_FACTORY = new NamedThreadFactory("time-series-metrics-publisher");

    private final Map<String, String> customTags;

    private final Map<String, DataType> tagDefine;

    public TimeSeriesMeterRegistry(TimeSeriesManager timeSeriesManager,
                                   TimeSeriesMetric metric,
                                   TimeSeriesRegistryProperties config,
                                   Map<String, String> customTags,
                                   String... tagKeys) {
        this(timeSeriesManager,
             metric,
             config,
             customTags,
             Arrays
                 .stream(tagKeys)
                 .collect(Collectors.toMap(Function.identity(), ignore -> StringType.GLOBAL)));
    }

    public TimeSeriesMeterRegistry(TimeSeriesManager timeSeriesManager,
                                   TimeSeriesMetric metric,
                                   TimeSeriesRegistryProperties config,
                                   Map<String, String> customTags,
                                   Map<String, DataType> tagDefine) {
        super(new TimeSeriesPropertiesPropertiesConfigAdapter(config), Clock.SYSTEM);
        this.timeSeriesManager = timeSeriesManager;
        this.metric = metric;
        this.customTags = customTags;

        this.tagDefine = new HashMap<>(tagDefine);

        start(DEFAULT_THREAD_FACTORY);
    }

    public void reload(){
        timeSeriesManager
            .registerMetadata(MeterTimeSeriesMetadata.of(metric, tagDefine))
            .doOnError(e -> log.error("register metric[{}] metadata error", metric.getId(), e))
            .subscribe((r) -> log.debug("register metric [{}] metadata success", metric.getId()));
    }

    @Override
    public void start(ThreadFactory threadFactory) {
        super.start(threadFactory);
        reload();
    }

    @Override
    protected void publish() {
        timeSeriesManager
            .getService(metric)
            .commit(Flux.fromIterable(this.getMeters())
                        .map(meter -> MeterTimeSeriesData.of(meter)
                                                         .name(getConventionName(meter.getId()))
                                                         .write(customTags)
                                                         .write(getConventionTags(meter.getId()), (key) -> tagDefine.getOrDefault(key, StringType.GLOBAL))))
            .doOnError(e -> log.error("failed to send metrics [{}]", metric.getId(), e))
//            .doOnSuccess(nil -> log.debug("success send metrics [{}]", metric.getId()))
            .subscribe();
    }


    @Override
    @Nonnull
    protected TimeUnit getBaseTimeUnit() {
        return TimeUnit.MILLISECONDS;
    }
}
